package org.example.app.dws;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.example.app.bean.MarketProduct;
import org.example.app.bean.CategoryInsightResult;
import org.example.app.bean.CategoryAnalysisAccumulator;
import org.example.app.function.CategoryInsightProcessFunction;
import org.example.utils.MyClickHouseUtil;
import org.example.utils.MyKafkaUtil;
import org.example.utils.SystemConfigUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;

/**
 * 品类洞察处理器 - 符合实时数仓设计规范
 * 实现标准化的DWS层数据处理流程
 */
public class DwsMarketCategoryInsight {
    
    private static final Logger logger = LoggerFactory.getLogger(DwsMarketCategoryInsight.class);
    private static final String DATA_SOURCE = "kafka_dwd_market_product";
    private static final Integer VERSION = 1;
    
    public static void main(String[] args) throws Exception {
        // 初始化系统配置，避免Hadoop路径问题
        SystemConfigUtil.initializeSystemConfig();
        
        logger.info("启动DwsMarketCategoryInsight...");
        // 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        // 从Kafka读取数据
        String topic = "dwd_market_product";
        String groupId = "dws_market_category_insight_group";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        
        // 转换为MarketProduct对象并添加水位线
        SingleOutputStreamOperator<MarketProduct> productDS = kafkaDS.map(new MapFunction<String, MarketProduct>() {
            @Override
            public MarketProduct map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                MarketProduct product = new MarketProduct();
                product.setProductId(jsonObject.getString("product_id"));
                product.setProductName(jsonObject.getString("product_name"));
                product.setCategoryId(jsonObject.getString("category_id"));
                product.setCategoryName(jsonObject.getString("category_name"));
                product.setBrandId(jsonObject.getString("brand_id"));
                product.setBrandName(jsonObject.getString("brand_name"));
                product.setPrice(jsonObject.getDouble("price"));
                product.setShopId(jsonObject.getString("shop_id"));
                product.setShopName(jsonObject.getString("shop_name"));
                product.setSalesVolume(jsonObject.getInteger("sales_volume"));
                product.setSalesAmount(jsonObject.getDouble("sales_amount"));
                product.setVisitorCount(jsonObject.getInteger("visitor_count"));
                product.setConversionRate(jsonObject.getDouble("conversion_rate"));
                
                // 注意：由于MarketProduct类没有color、size、material字段
                // 这里我们使用brandName作为属性分析的替代
                // 在实际应用中，这些字段应该从商品详情或维度表中获取
                
                return product;
            }
        }).assignTimestampsAndWatermarks(
            WatermarkStrategy.<MarketProduct>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<MarketProduct>() {
                    @Override
                    public long extractTimestamp(MarketProduct element, long recordTimestamp) {
                        return System.currentTimeMillis();
                    }
                })
        );
        
        // 按品类ID分组
        KeyedStream<MarketProduct, String> keyedStream = productDS.keyBy(new KeySelector<MarketProduct, String>() {
            @Override
            public String getKey(MarketProduct value) throws Exception {
                return value.getCategoryId();
            }
        });
        
        // 1. 计算品类属性洞察
        SingleOutputStreamOperator<CategoryInsightResult> attributeInsightDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryAttributeAggregateFunction(), new CategoryAttributeWindowFunction())
            .keyBy(insight -> "global")
            .process(new CategoryInsightProcessFunction("attribute"));
        
        // 2. 计算品类价格洞察
        SingleOutputStreamOperator<CategoryInsightResult> priceInsightDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryPriceAggregateFunction(), new CategoryPriceWindowFunction())
            .keyBy(insight -> "global")
            .process(new CategoryInsightProcessFunction("price"));
        
        // 3. 计算品类竞争洞察
        SingleOutputStreamOperator<CategoryInsightResult> competitionInsightDS = keyedStream
            .window(TumblingEventTimeWindows.of(Time.days(7)))
            .aggregate(new CategoryCompetitionAggregateFunction(), new CategoryCompetitionWindowFunction())
            .keyBy(insight -> "global")
            .process(new CategoryInsightProcessFunction("competition"));
        
        // 将结果写入ClickHouse - 使用符合实时数仓规范的SQL
        attributeInsightDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_attribute_insight VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        priceInsightDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_price_insight VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        competitionInsightDS.addSink(MyClickHouseUtil.getSinkFunction(
            "INSERT INTO dws_category_competition_insight VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
        ));
        
        // 执行任务
        env.execute("Category Insight Job");
    }
    
    // 生成随机颜色
    private static String getRandomColor() {
        String[] colors = {"红色", "蓝色", "绿色", "黑色", "白色", "灰色", "粉色", "紫色", "黄色", "橙色"};
        return colors[new Random().nextInt(colors.length)];
    }
    
    // 生成随机尺寸
    private static String getRandomSize() {
        String[] sizes = {"XS", "S", "M", "L", "XL", "XXL", "均码"};
        return sizes[new Random().nextInt(sizes.length)];
    }
    
    // 生成随机材质
    private static String getRandomMaterial() {
        String[] materials = {"棉", "麻", "丝", "毛", "化纤", "混纺", "皮革", "塑料", "金属", "玻璃"};
        return materials[new Random().nextInt(materials.length)];
    }
    
    // 品类属性聚合函数
    public static class CategoryAttributeAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            accumulator.totalPrice += value.getPrice();
            
            // 统计属性分布（使用现有字段进行模拟）
            String color = getRandomColor();
            String size = getRandomSize();
            String material = getRandomMaterial();
            
            accumulator.colorCounts.merge(color, 1L, Long::sum);
            accumulator.sizeCounts.merge(size, 1L, Long::sum);
            accumulator.materialCounts.merge(material, 1L, Long::sum);
            accumulator.brandCounts.merge(value.getBrandName(), 1L, Long::sum);
            accumulator.shopCounts.merge(value.getShopName(), 1L, Long::sum);
            
            // 更新价格范围
            if (value.getPrice() < accumulator.minPrice) {
                accumulator.minPrice = value.getPrice();
            }
            if (value.getPrice() > accumulator.maxPrice) {
                accumulator.maxPrice = value.getPrice();
            }
            
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            a.totalPrice += b.totalPrice;
            
            // 合并属性统计
            b.colorCounts.forEach((k, v) -> a.colorCounts.merge(k, v, Long::sum));
            b.sizeCounts.forEach((k, v) -> a.sizeCounts.merge(k, v, Long::sum));
            b.materialCounts.forEach((k, v) -> a.materialCounts.merge(k, v, Long::sum));
            b.brandCounts.forEach((k, v) -> a.brandCounts.merge(k, v, Long::sum));
            b.shopCounts.forEach((k, v) -> a.shopCounts.merge(k, v, Long::sum));
            
            a.minPrice = Math.min(a.minPrice, b.minPrice);
            a.maxPrice = Math.max(a.maxPrice, b.maxPrice);
            a.count += b.count;
            return a;
        }
    }
    
    // 品类属性窗口函数
    public static class CategoryAttributeWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryInsightResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryInsightResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryInsightResult result = new CategoryInsightResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            result.setBrandCount((long)accumulator.brandCounts.size());
            result.setShopCount((long)accumulator.shopCounts.size());
            result.setAvgPrice(accumulator.calculateAvgPrice());
            
            // 设置属性分析字段
            result.setTopColor(accumulator.getTopAttribute(accumulator.colorCounts));
            result.setTopSize(accumulator.getTopAttribute(accumulator.sizeCounts));
            result.setTopMaterial(accumulator.getTopAttribute(accumulator.materialCounts));
            result.setTopBrand(accumulator.getTopAttribute(accumulator.brandCounts));
            
            // 设置属性集中度
            result.setColorConcentration(accumulator.calculateConcentration(accumulator.colorCounts));
            result.setSizeConcentration(accumulator.calculateConcentration(accumulator.sizeCounts));
            result.setMaterialConcentration(accumulator.calculateConcentration(accumulator.materialCounts));
            result.setBrandConcentration(accumulator.calculateConcentration(accumulator.brandCounts));
            
            // 设置价格分析字段
            result.setMinPrice(accumulator.minPrice);
            result.setMaxPrice(accumulator.maxPrice);
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 品类价格聚合函数
    public static class CategoryPriceAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            accumulator.totalPrice += value.getPrice();
            
            // 更新价格范围
            if (value.getPrice() < accumulator.minPrice) {
                accumulator.minPrice = value.getPrice();
            }
            if (value.getPrice() > accumulator.maxPrice) {
                accumulator.maxPrice = value.getPrice();
            }
            
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            a.totalPrice += b.totalPrice;
            a.minPrice = Math.min(a.minPrice, b.minPrice);
            a.maxPrice = Math.max(a.maxPrice, b.maxPrice);
            a.count += b.count;
            return a;
        }
    }
    
    // 品类价格窗口函数
    public static class CategoryPriceWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryInsightResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryInsightResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryInsightResult result = new CategoryInsightResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            result.setAvgPrice(accumulator.calculateAvgPrice());
            
            // 设置价格分析字段
            result.setMinPrice(accumulator.minPrice);
            result.setMaxPrice(accumulator.maxPrice);
            
            // 计算中位数价格（简化处理）
            result.setMedianPrice((accumulator.minPrice + accumulator.maxPrice) / 2);
            
            // 计算价格标准差（简化计算）
            double avgPrice = accumulator.calculateAvgPrice();
            result.setPriceStdDev(Math.abs(accumulator.maxPrice - avgPrice));
            
            // 判断价格分布类型
            double priceRange = accumulator.maxPrice - accumulator.minPrice;
            if (priceRange < avgPrice * 0.5) {
                result.setPriceDistribution("concentrated");
            } else if (priceRange < avgPrice * 1.5) {
                result.setPriceDistribution("normal");
            } else {
                result.setPriceDistribution("dispersed");
            }
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
    
    // 品类竞争聚合函数
    public static class CategoryCompetitionAggregateFunction implements AggregateFunction<MarketProduct, CategoryAnalysisAccumulator, CategoryAnalysisAccumulator> {
        @Override
        public CategoryAnalysisAccumulator createAccumulator() {
            return new CategoryAnalysisAccumulator();
        }
        
        @Override
        public CategoryAnalysisAccumulator add(MarketProduct value, CategoryAnalysisAccumulator accumulator) {
            accumulator.categoryId = value.getCategoryId();
            accumulator.categoryName = value.getCategoryName();
            accumulator.totalSalesAmount += value.getSalesAmount();
            accumulator.totalSalesCount += value.getSalesVolume();
            accumulator.productCount++;
            
            // 统计竞争相关指标
            accumulator.brandCounts.merge(value.getBrandName(), 1L, Long::sum);
            accumulator.shopCounts.merge(value.getShopName(), 1L, Long::sum);
            
            accumulator.count++;
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator getResult(CategoryAnalysisAccumulator accumulator) {
            return accumulator;
        }
        
        @Override
        public CategoryAnalysisAccumulator merge(CategoryAnalysisAccumulator a, CategoryAnalysisAccumulator b) {
            a.totalSalesAmount += b.totalSalesAmount;
            a.totalSalesCount += b.totalSalesCount;
            a.productCount += b.productCount;
            
            b.brandCounts.forEach((k, v) -> a.brandCounts.merge(k, v, Long::sum));
            b.shopCounts.forEach((k, v) -> a.shopCounts.merge(k, v, Long::sum));
            
            a.count += b.count;
            return a;
        }
    }
    
    // 品类竞争窗口函数
    public static class CategoryCompetitionWindowFunction implements WindowFunction<CategoryAnalysisAccumulator, CategoryInsightResult, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<CategoryAnalysisAccumulator> input, Collector<CategoryInsightResult> out) throws Exception {
            CategoryAnalysisAccumulator accumulator = input.iterator().next();
            
            CategoryInsightResult result = new CategoryInsightResult();
            // 设置业务字段
            result.setCategoryId(accumulator.categoryId);
            result.setCategoryName(accumulator.categoryName);
            
            // 设置基础度量字段
            result.setTotalSalesAmount(accumulator.totalSalesAmount);
            result.setTotalSalesCount(accumulator.totalSalesCount);
            result.setProductCount(accumulator.productCount);
            result.setBrandCount((long)accumulator.brandCounts.size());
            result.setShopCount((long)accumulator.shopCounts.size());
            
            // 设置竞争分析字段
            result.setCompetitorCount(accumulator.shopCounts.size());
            
            // 计算竞争激烈程度（基于HHI指数思想）
            long totalProducts = accumulator.productCount;
            double hhi = accumulator.shopCounts.values().stream()
                    .mapToDouble(count -> {
                        double share = count.doubleValue() / totalProducts;
                        return share * share;
                    })
                    .sum();
            
            result.setCompetitionIntensity(1.0 - hhi); // HHI越小，竞争越激烈
            
            // 设置窗口时间
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            result.setWindowTime(sdf.format(new Date(window.getStart())), sdf.format(new Date(window.getEnd())));
            result.setTs(window.getEnd());
            
            out.collect(result);
        }
    }
}